package com.zjp.mg.config;

import com.mongodb.client.model.changestream.FullDocument;
import com.zjp.mg.convert.BigDecimalToDecimal128Converter;
import com.zjp.mg.convert.Decimal128ToBigDecimalConverter;
import com.zjp.mg.convert.ObjectIdToStringConverter;
import com.zjp.mg.convert.StringToObjectIdConverter;
import com.zjp.mg.core.OrdinaryDynamicMongoDBDataBaseImpl;
import com.zjp.mg.core.RequestDynamicMongoDBDataBaseImpl;
import org.bson.Document;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.MongoTransactionManager;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * @author 冷风
 */
@Component
public class MongodbConfig {

    @Bean
    public MongoTransactionManager mongoTransactionManager(MongoDatabaseFactory dbFactory) {
        return new MongoTransactionManager(dbFactory);
    }

    @Bean
    public BigDecimalToDecimal128Converter bigDecimalToDecimal128Converter() {
        return new BigDecimalToDecimal128Converter();
    }

    @Bean
    public Decimal128ToBigDecimalConverter decimal128ToBigDecimalConverter() {
        return new Decimal128ToBigDecimalConverter();
    }

    @Bean
    public ObjectIdToStringConverter objectIdToStringConverter() {
        return new ObjectIdToStringConverter();
    }

    @Bean
    public StringToObjectIdConverter stringToObjectIdConverter() {
        return new StringToObjectIdConverter();
    }


    // 复制集 使用 @Bean

    public MessageListenerContainer messageListenerContainer(MongoTemplate mongoTemplate,
                                                             DocumentMessageListener documentMessageListener) {

        Executor executor = Executors.newFixedThreadPool(5);


        MessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer(mongoTemplate, executor) {
            @Override
            public boolean isAutoStartup() {
                return true;
            }
        };
        ChangeStreamRequest<Document> request = ChangeStreamRequest.builder(documentMessageListener)
                .collection("user")  //需要监听的集合名
                //过滤需要监听的操作类型，可以根据需求指定过滤条件
                .filter(Aggregation.newAggregation(Aggregation.match(
                        Criteria.where("operationType").in("insert", "update", "delete"))))
                //不设置时，文档更新时，只会发送变更字段的信息，设置UPDATE_LOOKUP会返回文档的全部信息
                .fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
                .build();
        messageListenerContainer.register(request, Document.class);

        return messageListenerContainer;
    }


    @ConditionalOnProperty(prefix = "data.property", name = "open", havingValue = "false", matchIfMissing = true)
    @Bean
    public OrdinaryDynamicMongoDBDataBaseImpl ordinaryDynamicMongoDBDataBase(MongoTemplate mongoTemplate) {
        return new OrdinaryDynamicMongoDBDataBaseImpl(mongoTemplate);
    }


    @ConditionalOnProperty(prefix = "data.property", name = "open", havingValue = "true")
    @Bean
    public RequestDynamicMongoDBDataBaseImpl requestDynamicMongoDBDataBase(ConnectionConfiguration connectionConfiguration) {
        return new RequestDynamicMongoDBDataBaseImpl(connectionConfiguration);
    }
}
